﻿//PlazaStreamManager.cs
//Copyright (c) 2013 StockSharp LLC, all rights reserved.
//This code module is part of StockSharp library.
//This code is licensed under the GNU GENERAL PUBLIC LICENSE Version 3.
//See the file License.txt for the license details.
//More info on: http://stocksharp.com

namespace StockSharp.Plaza
{
	using System;
	using System.Collections.Generic;
	using System.IO;
	using System.Linq;

	using Ecng.Collections;
	using Ecng.Common;

	using StockSharp.Plaza.Metadata;
	using StockSharp.Plaza.Wrappers;

	/// <summary>
	/// Менеджер потоков данных.
	/// </summary>
	public class PlazaStreamManager : Disposable
	{
		private readonly PlazaTableSerializer _configSerializer = new PlazaTableSerializer();

		private readonly PlazaConnectionPool _connectionPool;

		private readonly SynchronizedList<PlazaConnection> _connections = new SynchronizedList<PlazaConnection>();

		internal PlazaStreamManager(PlazaConnectionPool connectionPool, Action<Exception> errorHandler)
		{
			if (connectionPool == null)
				throw new ArgumentNullException("connectionPool");

			_connectionPool = connectionPool;

			ConnectionGroupping = new Dictionary<PlazaStream, int>();
			RevisionManager = new PlazaRevisionManager(errorHandler);

			Streams = Enumerable.Empty<PlazaStream>();

			ConfigPath = "StockSharp_Plaza\\Configs";
		}

		private string _configPath;

		/// <summary>
		/// Путь к директории, в которой будут храниться схемы потоков Plaza.
		/// </summary>
		/// <remarks>
		/// По-умолчанию путь равен директории Configs, которая будет создана в директории StockSharp_Plaza.
		/// </remarks>
		public string ConfigPath
		{
			get { return _configPath; }
			set
			{
				if (ConfigPath == value)
					return;

				_configPath = value;

				if (!value.IsEmpty())
					Directory.CreateDirectory(value);
			}
		}

		/// <summary>
		/// Активные потоки.
		/// </summary>
		public IEnumerable<PlazaStream> Streams { get; private set; }

		/// <summary>
		/// Менеджер управления ревизиями потоков <see cref="PlazaStreamManager"/>.
		/// </summary>
		/// <remarks>Используется только в случае "безбазового" клиента.</remarks>
		public PlazaRevisionManager RevisionManager { get; private set; }

		/// <summary>
		/// Строка подключения к локальной базе данных, в которой хранятся скаченные с биржи данные.
		/// </summary>
		/// <remarks>Используется только в случае "базового" клиента.</remarks>
		public string ConnectionString { get; set; }

		/// <summary>
		/// Настройки группировки потоков по соединениям.
		/// </summary>
		public IDictionary<PlazaStream, int> ConnectionGroupping { get; private set; }

		/// <summary>
		/// Проверить, находятся ли все запущенные потоки в состоянии получении онлайн-данных.
		/// </summary>
		/// <returns>
		/// Находятся ли все запущенные потоки в состоянии получении онлайн-данных.
		/// </returns>
		public bool IsOnline()
		{
			return Streams.All(stream => stream.IsOnline());
		}

		internal bool IsExportStarted
		{
			get
			{
				lock (_connections.SyncRoot)
					return !_connections.IsEmpty();
			}
		}

		internal void StartExport(IDictionary<PlazaStream, PlazaTable[]> tables)
		{
			lock (_connections.SyncRoot)
			{
				if (IsExportStarted)
				{
					if (_connections.Any(c => c.NativeConnection.Status != ConnectionWrapperStatus.Disconnected))
						throw new InvalidOperationException("Потоки данных не были остановлены с предыдущего запуска.");
					else
						_connections.Clear();
				}

				Streams = tables.Keys.ToArray();

				foreach (var pair in tables)
				{
					var stream = pair.Key;
					var streamTables = pair.Value;

					if (ConfigPath.IsEmpty())
					{
						var tempFileName = Path.GetTempFileName();
						stream.ConfigFileName = tempFileName.Replace("tmp", "ini");
						File.Delete(tempFileName);
					}
					else
					{
						stream.ConfigFileName = Path.Combine(ConfigPath, "{0}.ini".Put(stream.Name));
					}

					_configSerializer.Serialize(stream, streamTables);
				}

				if (!RevisionManager.Tables.IsEmpty())
				{
					if (!ConnectionString.IsEmpty())
						throw new InvalidOperationException("Одновременно используются механизмы базового и безбазового клиентов.");

					RevisionManager.Start(tables);
				}

				var connections = new List<PlazaConnection>();

				foreach (var group in Streams.GroupBy(s => ConnectionGroupping.TryGetValue2(s) ?? -1).ToArray())
				{
					var grouppedStreams = group.Select(g => new RefTriple<PlazaStream, PlazaTable[], int>(g, tables[g], 0)).ToArray();

					connections.Add(_connectionPool.RunConnection("Plaza. Stream thread.",
					connection =>
					{
						foreach (var tuple in grouppedStreams)
						{
							var stream = tuple.First;

							stream.RaiseLog("Инициализация {0} потока.".Put(stream));

							stream.Init();

							if (!ConnectionString.IsEmpty())
								stream.NativeStream.ConnectionString = ConnectionString;

							tuple.Second.ForEach(t => t.Subscribe(stream));
						}
					},
					(connection, processError) =>
					{
						if (connection.NativeConnection.Status != (ConnectionWrapperStatus.Connected | ConnectionWrapperStatus.RouterConnected))
						{
							foreach (var tuple in grouppedStreams)
							{
								var stream = tuple.First;

								try
								{
									stream.RaiseLog("Закрытие {0} потока.".Put(stream));
									stream.Close();
								}
								catch (Exception ex)
								{
									processError(GetStreamException(stream, ex));
								}
							}
						}
						else
						{
							foreach (var tuple in grouppedStreams)
							{
								var stream = tuple.First;

								try
								{
									if (tuple.Third > 0)
									{
										tuple.Third--;
										continue;
									}

									var streamTables = tuple.Second;

									var nativeStream = stream.NativeStream;

									if (nativeStream.State == StreamWrapperStates.Error || nativeStream.State == StreamWrapperStates.Closed)
									{
										stream.RaiseLog("Состояние потока {0}.".Put(nativeStream.State));

										if (nativeStream.State == StreamWrapperStates.Error)
											nativeStream.Close();

										RevisionManager.LoadState(stream);

										foreach (var table in streamTables)
											nativeStream.TableSet.UpdateRevision(table.Name, table.Revision);

										nativeStream.Open(connection.NativeConnection);

										stream.RaiseLog("Поток {0} переоткрыт.".Put(stream));
									}
									else
									{
										RevisionManager.SaveState(stream);
									}
								}
								catch (Exception ex)
								{
									processError(GetStreamException(stream, ex));

									// для потока выбросившего ошибку пропускаем 1000 обращений, чтобы не приводить к постоянным логам с ошибками
									tuple.Third = 1000;
								}
							}
						}
					},
					connection => grouppedStreams.ForEach(tuple =>
					{
						var stream = tuple.First;
						var streamTables = tuple.Second;

						stream.RaiseLog("Остановка потока {0}.".Put(stream));

						streamTables.ForEach(t => t.UnSubscribe(stream));

						stream.Close();
						stream.CloseNativeStream();

						stream.RaiseLog("Поток {0} остановлен.".Put(stream));
					})));
				}

				foreach (var connection in connections)
				{
					connection.Start();
					_connections.Add(connection);
				}
			}
		}

		internal void StopExport()
		{
			lock (_connections.SyncRoot)
				_connections.ForEach(c => c.Stop(() => _connections.Remove(c)));

			RevisionManager.Stop();
		}

		private static Exception GetStreamException(PlazaStream stream, Exception innerException)
		{
			if (stream == null)
				throw new ArgumentNullException("stream");

			if (innerException == null)
				throw new ArgumentNullException("innerException");

			return new InvalidOperationException("Поток {0} выбросил ошибку.".Put(stream), innerException);
		}

		/// <summary>
		/// Освободить занятые ресурсы.
		/// </summary>
		protected override void DisposeManaged()
		{
			StopExport();
			RevisionManager.Dispose();
			base.DisposeManaged();
		}
	}
}
